package cn.me.alphamq.client.producer;

import cn.me.alphamq.common.remote.DefaultClient;
import cn.me.alphamq.common.store.TxListenerManager;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.*;
import cn.me.alphamq.common.entity.resp.PushDelayMsgResp;
import cn.me.alphamq.common.entity.resp.PushMsgResp;
import cn.me.alphamq.common.entity.resp.PushTxMsgResp;
import cn.me.alphamq.common.enumeration.*;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.remote.NettyClient;
import cn.me.alphamq.common.transaction.TransactionListener;
import cn.me.alphamq.common.util.AssertUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * 基础的生产者类
 *
 * @author f
 */
public class DefaultProducer extends DefaultClient implements Producer {

    private static final Logger log = LoggerFactory.getLogger(DefaultProducer.class);

    /**
     * 能够唯一标识生产者的 id
     */
    private final String producerId;

    /**
     * 本生产者的事务监听器
     */
    private TransactionListener transactionListener;

    @Override
    public SendResult sendSync(Message message) {
        NettyClient brokerClient = pickBrokerClient();

        TcpMsg resp = brokerClient.sendMsgSync(EventCodeEnum.PUSH_MSG_REQ, new PushMsgReq(
                message.getTopic(), message.getQueueId(), message.getBody()));
        if (resp == null) {
            log.error("sendSync fail: broker disconnect!");
            return new SendResult(PushStatusEnum.FAIL.getCode());
        }

        PushMsgResp pushMsgResp = JSON.parseObject(resp.getBody(), PushMsgResp.class);
        int pushStatus = pushMsgResp.getPushStatus();
        if (PushStatusEnum.FAIL.getCode() == pushStatus) {
            log.error("sendSync fail: {}", pushMsgResp.getDesc());
        }
        return new SendResult(pushStatus);
    }

    @Override
    public void sendAsync(Message message) {
        pickBrokerClient().sendMsgAsync(EventCodeEnum.PUSH_MSG_REQ, new PushMsgReq(
                message.getTopic(), message.getQueueId(), message.getBody()));
    }

    @Override
    public SendResult sendDelaySync(Message message, int delayTime) {
        NettyClient brokerClient = pickBrokerClient();

        TcpMsg resp = brokerClient.sendMsgSync(EventCodeEnum.PUSH_DELAY_MSG_REQ, new PushDelayMsgReq(
                message.getTopic(), message.getQueueId(), message.getBody(), delayTime));
        if (resp == null) {
            log.error("sendDelaySync fail: broker disconnect!");
            return new SendResult(PushStatusEnum.FAIL.getCode());
        }

        PushDelayMsgResp pushDelayMsgResp = JSON.parseObject(resp.getBody(), PushDelayMsgResp.class);
        int pushStatus = pushDelayMsgResp.getPushStatus();
        if (PushStatusEnum.FAIL.getCode() == pushStatus) {
            log.error("sendDelaySync fail: {}", pushDelayMsgResp.getDesc());
        }
        return new SendResult(pushStatus);
    }

    @Override
    public void sendDelayAsync(Message message, int delayTime) {
        pickBrokerClient().sendMsgAsync(EventCodeEnum.PUSH_DELAY_MSG_REQ, new PushDelayMsgReq(
                message.getTopic(), message.getQueueId(), message.getBody(), delayTime));
    }

    @Override
    public SendResult sendTxSync(Message message) {
        AssertUtil.isNotNull(transactionListener, "transactionListener is null");

        NettyClient nettyClient = pickBrokerClient();

        String msgId = UUID.randomUUID().toString();
        TcpMsg halfResp = nettyClient.sendMsgSync(EventCodeEnum.PUSH_TX_MSG_REQ, msgId,
                PushTxMsgReq.newHalfMsg(message, msgId, producerId));
        if (halfResp == null) {
            log.error("sendTxSync half fail: broker disconnect!");
            return new SendResult(PushStatusEnum.FAIL.getCode());
        }

        PushTxMsgResp pushTxMsgResp = JSON.parseObject(halfResp.getBody(), PushTxMsgResp.class);
        int pushStatus = pushTxMsgResp.getPushStatus();
        if (PushStatusEnum.FAIL.getCode() == pushStatus) {
            log.error("sendTxSync half fail: {}", pushTxMsgResp.getDesc());
            return new SendResult(PushStatusEnum.FAIL.getCode());
        }

        // 执行本地事务
        LocalTxStateEnum localTxState = transactionListener.executeLocalTransaction(message);
        PushTxMsgReq pushTxMsgReq = PushTxMsgReq.newRemainMsg(message, msgId, producerId, localTxState);
        if (LocalTxStateEnum.COMMIT.equals(localTxState)) {
            // 如果是已提交状态
            TcpMsg commitResp = nettyClient.sendMsgSync(EventCodeEnum.PUSH_TX_MSG_REQ, msgId, pushTxMsgReq);
        } else if (LocalTxStateEnum.ROLLBACK.equals(localTxState)) {
            // 如果是回滚状态
            TcpMsg rollbackResp = nettyClient.sendMsgSync(EventCodeEnum.PUSH_TX_MSG_REQ, msgId, pushTxMsgReq);
        } else if (LocalTxStateEnum.UNKNOWN.equals(localTxState)) {
            // 如果是未知状态，则等待 broker 回调检查本地事务状态
        }

        return new SendResult(pushStatus);
    }

    public DefaultProducer(String nameServerIp, int nameServerPort,
                           String username, String password,
                           String brokerGroup) {
        super(BrokerRoleEnum.MASTER.getCode(), brokerGroup, nameServerIp, nameServerPort,
                "default-producer-nameserver-client");

        this.producerId = UUID.randomUUID().toString();

        super.start(username, password, ServiceInstanceTypeEnum.PRODUCER);
    }

    /**
     * 设置事务监听器，并将其放入管理者中进行管理
     *
     * @param transactionListener 待设置的事务监听器
     */
    public void setTransactionListener(TransactionListener transactionListener) {
        this.transactionListener = transactionListener;
        TxListenerManager.put(producerId, transactionListener);
    }

    @Override
    protected String genBrokerClientName(String brokerAddress) {
        return "default-producer-broker-client-on[" + brokerAddress + "]";
    }
}
